package edu.hawaii.ics.pjdb.worker.command_runners;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;

import org.apache.thrift.TException;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import edu.hawaii.ics.pjdb.result_transport.TupleListener;
import edu.hawaii.ics.pjdb.result_transport.TupleServiceHelper;
import edu.hawaii.ics.pjdb.types.DbEnrichCommand;
import edu.hawaii.ics.pjdb.types.Tuple;
import edu.hawaii.ics.pjdb.types.TupleService;
import edu.hawaii.ics.pjdb.types.WorkerCommand;

public class DbEnrichCommandRunner extends CommandRunner {
    private static final Logger logger = LoggerFactory
            .getLogger(DbEnrichCommandRunner.class);

    protected static class EnrichTupleListener extends TupleListener {
        private static final Logger logger = LoggerFactory
                .getLogger(EnrichTupleListener.class);

        private final DbEnrichCommand cmd;
        private final Connection localDb;
        private final TupleService.Client out;

        public EnrichTupleListener(WorkerCommand workerCmd, Connection localDb)
                throws SQLException, TTransportException, InterruptedException {
            this.localDb = localDb;
            this.cmd = workerCmd.getCommand().getDbEnrichCmd();
            this.out = TupleServiceHelper.createClient(workerCmd
                    .getOutputAddress());

            logger.info(
                    "Stuff is all set up. Listening for tuples, sending enriched ones to {}",
                    workerCmd.getOutputAddress());
        }

        @Override
        public void receiveTuples(List<Tuple> tuples) {
            logger.debug("#receiveTuples() called.");

            try {
                PreparedStatement statement = localDb.prepareStatement(cmd
                        .getSqlStatement());
                for (Tuple tuple : tuples) {
                    String fieldValue = tuple.getEntries().get(
                            cmd.getJoinFieldIndex());
                    logger.info("Received tuple: {}, field value = {}",
                            tuple.getEntries(), fieldValue);
                    statement.setObject(1, fieldValue);
                    ResultSet results = statement.executeQuery();
                    TupleServiceHelper.submitTuples(out, results, tuple);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void finished() {
            logger.info("#finish() called.");

            try {
                out.finished();
            } catch (Exception e) {
                // We should expect this to happen.
            }
        }
    }

    public DbEnrichCommandRunner(Connection localDb) {
        super(localDb);
    }

    @Override
    public void execute(WorkerCommand cmd) throws TException, SQLException {
        logger.info(
                "Received dbenrich command. Setting up input listener on port {}",
                cmd.getCommand().getDbEnrichCmd().getInputAddress().getPort());

        try {
            EnrichTupleListener enrichWorker = new EnrichTupleListener(cmd,
                    localDb);
            TServer inputServer = TupleServiceHelper.createNonThreadedServer(
                    cmd.getCommand().getDbEnrichCmd().getInputAddress()
                            .getPort(), enrichWorker);
            inputServer.serve();
        } catch (InterruptedException e) {
            throw new TException(e);
        }
    }
}
